#ifndef __MQ_CONNECTION_HPP__
#define __MQ_CONNECTION_HPP__

#include "../mqthird/include/muduo/proto/dispatcher.h"
#include "../mqthird/include/muduo/proto/codec.h"
// #include "muduo/proto/query.pb.h"

#include "../include/muduo/base/Logging.h"
#include "../include/muduo/base/Mutex.h"
#include "../include/muduo/net/EventLoop.h"
#include "../include/muduo/net/TcpClient.h"

#include <stdio.h>
#include <unistd.h>
#include <iostream>
#include "../mqthird/include/muduo/net/TcpClient.h"
#include "../mqthird/include/muduo/net/EventLoopThread.h"
#include "../mqthird/include/muduo/net/TcpConnection.h"
#include "../mqthird/include/muduo/base/CountDownLatch.h"
#include <functional>
#include <string>
// #include "request.pb.h"

#include "mq_channel.hpp"
#include "mq_worker.hpp"

namespace mq
{
    class Connection
    {
    public:
        using Ptr = std::shared_ptr<Connection>;
        typedef std::shared_ptr<google::protobuf::Message> MessagePtr;

        Connection(const std::string& sip,int sport,AsyncWorker::Ptr& worker)
            :_latch(1)
            ,_client(worker->loopthread.startLoop(),muduo::net::InetAddress(sip,sport),"Connection")
            ,_dispatcher(std::bind(&Connection::onUnknownMessage,this,std::placeholders::_1
                ,std::placeholders::_2,std::placeholders::_3))
            ,_codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage,&_dispatcher,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3)))
            ,_channel_manager(std::make_shared<ChannelManager>())
            ,_worker(worker)
        {
            // 注册业务请求处理函数。
            _dispatcher.registerMessageCallback<mq::basicCommonResponse>(
            std::bind(&Connection::basicResponse, this, std::placeholders::_1
                ,std::placeholders:: _2
                ,std::placeholders:: _3));

            _dispatcher.registerMessageCallback<mq::basicConsumeResponse>(
            std::bind(&Connection::consumeResponse, this, std::placeholders::_1
                ,std::placeholders:: _2
                ,std::placeholders:: _3));

            _client.setConnectionCallback(
            std::bind(&Connection::onConnection, this, std::placeholders::_1));

            _client.setMessageCallback(
            std::bind(&ProtobufCodec::onMessage, _codec.get(), std::placeholders::_1
                ,std::placeholders:: _2
                ,std::placeholders:: _3));

            _client.connect();
            _latch.wait();
        }
        Channel::Ptr openChannel()
        {
            Channel::Ptr channel = _channel_manager->create(_conn,_codec);
            bool ret = channel->openChannel();
            
            if(ret == false)
            {
                DLOG("打开信道失败。");

                return Channel::Ptr();
            }

            return channel;
        }
        void closeChannel(Channel::Ptr channel)
        {
            channel->closeChannel();

            return ;
        }
    private: 
        void basicResponse(const muduo::net::TcpConnectionPtr& conn,
                const basicCommonResponsePtr& message,
                muduo::Timestamp)
        {
            // 1.找到信道
            Channel::Ptr channel = _channel_manager->get(message->cid());
            if(channel.get() == nullptr)
            {
                DLOG("未找到信道信息");

                return ;
            }

            // 2. 将得到的响应条件，添加到信道的基础响应hash_map中。
            channel->putBasicResponse(message);
            
            return ;
        }
        void consumeResponse(const muduo::net::TcpConnectionPtr conn
            ,const basicConsumeResponsePtr& message
            ,muduo::Timestamp)
        {
            // 1. 找到信道。
            Channel::Ptr channel = _channel_manager->get(message->cid());
            if(channel.get() == nullptr)
            {
                DLOG("未找到信道信息。");

                return ;
            }

            // 2. 封装异步任务。
            _worker->pool.push([channel,message]()
            {
                channel->consume(message);
            });
        }
        void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn
            ,const MessagePtr& message,muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }
        void onConnection(const muduo::net::TcpConnectionPtr& conn)
        {
            if(conn->connected() == true)
            {
                LOG_INFO << "新链接已建立" ;
                _latch.countDown(); // 唤醒主线程中的阻塞
                _conn = conn;
            }
            else
            {
                LOG_INFO << "链接即将关闭" ;
                _conn.reset();
            }

            return ;
        }
    private:
        muduo::CountDownLatch _latch;  // 实现同步
        // muduo::net::EventLoopThread _loopthread;  // 异步循环处理线程
        muduo::net::TcpConnectionPtr _conn; // 客户端对应的连接
        muduo::net::TcpClient _client;  // 客户端
        ProtobufDispatcher _dispatcher; // 请求分发器
        ProtobufCodecPtr _codec; // 协议处理器

        AsyncWorker::Ptr _worker;
        ChannelManager::Ptr _channel_manager;
    };
}

#endif
